package cn.doitedu.rtdw.dataetl;

import cn.doitedu.rtdw.udfs.MyTimeTruncUdf;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class JOB02_流量概况OLAP支撑 {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 建表，映射 kafka中 dwd层 行为日志明细宽表
        tenv.executeSql(
                " create table dwd_events_kafka (                       "+
                        "     release_channel string                             "+
                        "     ,device_type string                                 "+
                        "     ,session_id  string                                 "+
                        "     ,lat         double                                 "+
                        "     ,lng         double                                 "+
                        "     ,username    string                                 "+
                        "     ,event_id     string                                "+
                        "     ,action_time   bigint                               "+
                        "     ,properties  map<string,string>                     "+
                        "     ,user_id  bigint                                    "+
                        " 	  ,member_level_id bigint                             "+
                        "     ,password string                                    "+
                        "     ,nickname string                                    "+
                        "     ,phone string                                       "+
                        "     ,status int                                         "+
                        "     ,create_time timestamp(3)                           "+
                        "     ,icon string                                        "+
                        "     ,gender int                                         "+
                        "     ,birthday date                                      "+
                        "     ,register_city string                               "+
                        "     ,job string                                         "+
                        "     ,personalized_signature string                      "+
                        "     ,source_type int                                    "+
                        "     ,integration int                                    "+
                        "     ,growth int                                         "+
                        "     ,lucky_count int                                    "+
                        "     ,history_integration int                            "+
                        "     ,modify_time timestamp(3)                           "+
                        "     ,province string                                    "+
                        "     ,city string                                        "+
                        "     ,region string                                      "+
                        "     ,page_type string                                   "+
                        "     ,page_service string                                "+
                        "     ,page_lanmu string                                  "+
                        "     ,page_pindao string                                 "+
                        "     ,pt AS proctime()                                   "+   // 处理时间语义
                        "     ,rt AS to_timestamp_ltz(action_time,3)              "+
                        "     ,watermark for rt as rt - interval '0' second       "+   // 事件时间语义
                        " ) with (                                                "+
                        "     'connector' = 'kafka',                              "+
                        "     'topic' = 'dwd-events',                             "+
                        "     'properties.bootstrap.servers' = 'doitedu:9092',    "+
                        "     'properties.group.id' = 'goo2',                     "+
                        "     'scan.startup.mode' = 'latest-offset',              "+
                        "     'value.format'='json',                              "+
                        "     'value.json.fail-on-missing-field'='false',         "+
                        "     'value.fields-include' = 'EXCEPT_KEY'               "+
                        " )                                                       "
        );


        // 统计主题需求中的各种维度+user_id维度+session_id维度，下的 pv数
        tenv.createTemporaryFunction("my_time_trunc", MyTimeTruncUdf.class);
        tenv.executeSql(
                "CREATE TEMPORARY VIEW tmp_result AS " +
                        " WITH tmp AS (                                              "+
                        "     SELECT                                                 "+
                        "       user_id,                                           "+
                        " 		session_id,                                          "+
                        " 		province,                                            "+
                        " 		city,                                                "+
                        " 		region,                                              "+
                        " 		device_type,                                         "+
                        " 		release_channel,                                     "+
                        " 		page_type,                                           "+
                        " 		page_lanmu,                                          "+
                        " 		page_service,                                        "+
                        " 		to_date(date_format(to_timestamp_ltz(action_time,3),'yyyy-MM-dd')) as dt,         "+
                        " 		my_time_trunc(action_time,60) as time_60m,           "+
                        " 		my_time_trunc(action_time,30) as time_30m,           "+
                        " 		my_time_trunc(action_time,10) as time_10m,           "+
                        " 		my_time_trunc(action_time, 5) as time_05m,           "+
                        " 		my_time_trunc(action_time, 1) as time_01m,           "+
                        "       rt		                                             "+
                        "     FROM dwd_events_kafka                                  "+
                        "     WHERE event_id = 'page_load'                           "+
                        " )                                                          "+
                        "                                                            "+
                        " SELECT                                                     "+
                        " 	province,                                                "+
                        " 	city,                                                    "+
                        " 	region,                                                  "+
                        " 	device_type,                                             "+
                        " 	release_channel,                                         "+
                        " 	page_type,                                               "+
                        " 	page_lanmu,                                              "+
                        " 	page_service,                                            "+
                        " 	dt,                                                      "+
                        " 	time_60m,                                                "+
                        " 	time_30m,                                                "+
                        " 	time_10m,                                                "+
                        " 	time_05m,                                                "+
                        " 	time_01m,                                                "+
                        " 	COUNT(1) AS pv,                                          "+
                        "   user_id,                                                 "+
                        " 	session_id                                              "+
                        " FROM TABLE(                                                "+
                        "     TUMBLE(TABLE tmp,DESCRIPTOR(rt),INTERVAL '1' MINUTE)   "+
                        " )                                                          "+
                        " GROUP BY                                                   "+
                        "     window_start,                                          "+
                        "     window_end,	                                         "+
                        "     user_id,                                               "+
                        " 	session_id,                                              "+
                        " 	province,                                                "+
                        " 	city,                                                    "+
                        " 	region,                                                  "+
                        " 	device_type,                                             "+
                        " 	release_channel,                                         "+
                        " 	page_type,                                               "+
                        " 	page_lanmu,                                              "+
                        " 	page_service,                                            "+
                        " 	dt,                                                      "+
                        " 	time_60m,                                                "+
                        " 	time_30m,                                                "+
                        " 	time_10m,                                                "+
                        " 	time_05m,                                                "+
                        " 	time_01m                                                 "

        );


        // 创建doris物理表的逻辑映射表
        tenv.executeSql(
                " create table agg_doris(                      "+
                        "      province          varchar(20)            "+
                        "     ,city              varchar(20)            "+
                        "     ,region            varchar(20)            "+
                        "     ,device_type       varchar(20)            "+
                        "     ,release_channel   varchar(20)            "+
                        "     ,page_type         varchar(20)            "+
                        "     ,page_lanmu        varchar(20)            "+
                        "     ,page_service      varchar(20)            "+
                        "     ,dt                date                   "+
                        "     ,time_60m          varchar(20)            "+
                        "     ,time_30m          varchar(20)            "+
                        "     ,time_10m          varchar(20)            "+
                        "     ,time_05m          varchar(20)            "+
                        "     ,time_01m          varchar(20)            "+
                        " 	  ,pv      bigint                                                                                "+
                        " 	  ,uv      bigint                                                                                "+
                        " 	  ,ses_cnt string                                                                                "+
                        " ) WITH (                                                                                           "+
                        "    'connector' = 'doris',                                                                          "+
                        "    'fenodes' = 'doitedu:8030',                                                                     "+
                        "    'table.identifier' = 'dws.app_traffic_agg_01',                                                  "+
                        "    'username' = 'root',                                                                            "+
                        "    'password' = 'root',                                                                            "+
                        "    'sink.label-prefix' = 'doit42-02-2',                                                              "+
                        "    'sink.properties.columns' = 'province,city,region,device_type,release_channel,                  "+
                        "    page_type,page_lanmu,page_service,dt,time_60m,time_30m,time_10m,time_05m,time_01m,  "+
                        "    pv,uv,ses_cnt,uv=to_bitmap(uv),ses_cnt=hll_hash(ses_cnt)'                                       "+
                        " )                                                                                                  "
        );


        // 写一个insert语句
        tenv.executeSql("insert into agg_doris select * from tmp_result");

    }

}
